package com.gobrs.async.core;

import com.gobrs.async.core.common.domain.AsyncParam;
import com.gobrs.async.core.common.util.IdWorker;
import com.gobrs.async.core.common.util.JsonUtil;
import com.gobrs.async.core.common.util.SystemClock;
import com.gobrs.async.core.config.ConfigManager;
import com.gobrs.async.core.holder.BeanHolder;
import com.gobrs.async.core.log.LogWrapper;
import com.gobrs.async.core.log.TraceUtil;
import com.gobrs.async.core.task.AsyncTask;
import com.gobrs.async.core.threadpool.GobrsAsyncThreadPoolFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 任务触发器
 * 一条规则对应一个触发器
 * 负责任务执行图的构建：包括每个任务对应任务执行器，每个任务的上下直接父或子任务依赖关系
 */
public class TaskTrigger<P, R> {

    Logger logger = LoggerFactory.getLogger(TaskTrigger.class);

    private final TaskFlow taskFlow;

    private GobrsAsyncThreadPoolFactory threadPoolFactory = BeanHolder.getBean(GobrsAsyncThreadPoolFactory.class);

    /**
     * 待执行的任务集合 任务->任务执行器 (用于任务执行器执行时 读)
     */
    private Map<AsyncTask, TaskActuator> prepareTaskMap = Collections.synchronizedMap(new IdentityHashMap<>());

    /**
     * 待执行的任务集合 任务->任务执行器 (用于任务执行器构建时 写)
     */
    private Map<AsyncTask, TaskActuator> prepareTaskMapWrite = new IdentityHashMap<>();

    private String ruleName;

    /**
     * 作为 没有子任务的任务 的子任务
     */
    public AssistantTask assistantTask;

    /**
     * 任务 -> 任务前的直接父任务
     */
    public Map<AsyncTask, List<AsyncTask>> upwardTasksMapSpace = new ConcurrentHashMap<>();

    TaskTrigger(String ruleName, TaskFlow taskFlow) {
        this.ruleName = ruleName;
        this.taskFlow = taskFlow;
        prepare();
    }

    /**
     * 预加载任务依赖关系
     * 缓存依赖任务配置、维护 任务所依赖的任务数量和任务对象
     */
    private void prepare() {
        /**
         * 任务 -> 任务后的直接子任务
         */
        Map<AsyncTask, List<AsyncTask>> downTasksMap = copyDependTasks(taskFlow.getDependsTasks());
        /**
         * 任务 -> 任务前的直接父任务
         */
        Map<AsyncTask, List<AsyncTask>> upwardTasksMap = new HashMap<>();

        for (AsyncTask task : downTasksMap.keySet()) {
            upwardTasksMap.put(task, new ArrayList<>(1));
        }

        for (AsyncTask task : downTasksMap.keySet()) {
            for (AsyncTask depended : downTasksMap.get(task)) {
                upwardTasksMap.get(depended).add(task);
            }
        }

        assistantTask = new AssistantTask();

        /**
         * 没有任何子任务的任务
         */
        List<AsyncTask> noSubtasks = new ArrayList<>(1);

        // 给没有子任务的任务 附上 辅助任务
        for (AsyncTask task : downTasksMap.keySet()) {
            List<AsyncTask> dTasks = downTasksMap.get(task);
            if (dTasks.isEmpty()) {
                noSubtasks.add(task);
                downTasksMap.get(task).add(assistantTask);
            }
        }
        // 辅助任务的直接子任务为空集合
        downTasksMap.put(assistantTask, new ArrayList<>(0));
        // 辅助任务的直接父任务为 没有任何子任务的任务
        upwardTasksMap.put(assistantTask, noSubtasks);
        upwardTasksMapSpace = upwardTasksMap;
        clear();
        // 构建每个任务的 TaskActuator
        for (AsyncTask task : downTasksMap.keySet()) {
            TaskActuator process;
            if (task != assistantTask) {
                List<AsyncTask> circularDependency = upwardTasksMap.get(task).stream().filter(x -> x.getName().equals(task.getName())).collect(Collectors.toList());
                // task的直接父任务的个数
                int upDepend = 0;
                if (circularDependency.size() == 0) {
                    upDepend = upwardTasksMap.get(task).size();
                }
                /**
                 * 每个业务任务都使用一个新的任务执行器执行
                 */
                process = new TaskActuator(task, upDepend, downTasksMap.get(task), upwardTasksMap);
            } else {
                /***
                 * 终止任务
                 */
                process = new TerminationTask(task, upwardTasksMap.get(task).size(), downTasksMap.get(task));
            }
            prepareTaskMapWrite.put(task, process);
        }
        /**
         * 读写分离
         */
        prepareTaskMap = prepareTaskMapWrite;
        if (logger.isInfoEnabled()) {
            logger.info("prepareTaskMap build success {}", JsonUtil.obj2String(prepareTaskMap));
        }
    }

    /**
     * 清空写配置缓存
     */
    private void clear() {
        prepareTaskMapWrite.clear();
    }

    /**
     * 建立 任务 -> 任务下的子任务 映射
     * @param handlerMap
     * @return
     */
    private Map<AsyncTask, List<AsyncTask>> copyDependTasks(Map<AsyncTask, List<AsyncTask>> handlerMap) {
        IdentityHashMap<AsyncTask, List<AsyncTask>> rt = new IdentityHashMap<>();
        for (AsyncTask asyncTask : handlerMap.keySet()) {
            rt.put(asyncTask, new ArrayList<>(handlerMap.get(asyncTask)));
        }
        return rt;
    }

    /**
     * 触发任务加载
     */
    TaskLoader trigger(AsyncParam param, long timeout) {
        return trigger(param, timeout, null);
    }

    /**
     * 触发任务加载
     */
    TaskLoader trigger(AsyncParam param, Set<String> affirTasks, long timeout) {
        return trigger(param, timeout, affirTasks);
    }

    /**
     * 触发任务流并获取任务加载器
     * 链路日志
     * 线程池配置
     * 从缓存好的预加载配置中 clone副本 浅clone 只备份基本类型
     * @param param 任务参数
     * @param timeout 任务超时阈值
     * @param optionalTasks 可选任务
     * @return
     */
    TaskLoader trigger(AsyncParam<P> param, long timeout, Set<String> optionalTasks) {
        IdentityHashMap<AsyncTask, TaskActuator> newProcessMap = new IdentityHashMap<>(prepareTaskMap.size());
        /**
         * 创建一个任务加载器，一个规则对应一个任务加载器
         */
        TaskLoader<P, R> loader = new TaskLoader<P, R>(ruleName, threadPoolFactory.getThreadPoolExecutor(), newProcessMap, timeout);

        TaskSupport support = related(param, loader);

        for (AsyncTask task : prepareTaskMap.keySet()) {
            /**
             * 克隆用于线程隔离的任务执行器
             */
            TaskActuator processor = (TaskActuator) prepareTaskMap.get(task).clone();
            processor.init(support, param);
            newProcessMap.put(task, processor);
        }

        Optimal.doOptimal(optionalTasks, loader, upwardTasksMapSpace);
        return loader;
    }

    /**
     * 设置 任务总线和任务加载器关联关系
     * 配置设置 环境加载
     */
    private TaskSupport related(AsyncParam<P> param, TaskLoader<P, R> loader) {
        TaskSupport support = getSupport(param);
        support.setTaskLoader(loader);
        logAdvance(support);
        loader.setAssistantTask(assistantTask);

        /**
         * 从线程池工厂获取线程池
         */
        support.setExecutorService(threadPoolFactory.getThreadPoolExecutor());
        return support;
    }

    /**
     * 终止任务 在整个任务流程结束后 会调用该任务类的 completed() 方法
     * 任务流程结束任务
     * 作为 协助任务的任务执行器
     */
    private class TerminationTask<P, R> extends TaskActuator {

        /**
         * 任务执行者
         * @param handler       处理者
         * @param depdending    所依赖的任务数量
         * @param dependedTasks 依赖任务组
         */
        TerminationTask(AsyncTask handler, int depdending, List<AsyncTask> dependedTasks) {
            super(handler, depdending, dependedTasks);
        }

        /**
         * Task completion interrupt the main thread blocks
         */
        @Override
        public Object call() {
            support.taskLoader.completed();
            return null;
        }
    }

    /**
     * 助理任务 通过助理任务可以协助组装任务依赖配置
     */
    public class AssistantTask<P, R> extends AsyncTask<P, R> {
        @Override
        public R task(P p, TaskSupport support) {
            return null;
        }
    }

    private void doProcess(TaskLoader taskLoader, TaskActuator process, Set<AsyncTask> affirTasks) {
        if (affirTasks != null) {
            if (affirTasks.contains(process.getTask())) {
                taskLoader.oplCount.incrementAndGet();
                taskLoader.startProcess(process);
            }
        } else {
            taskLoader.startProcess(process);
        }
    }

    /**
     * 获取任务流程 总线
     * support 封装流程中重要的配置信息和任务执行过程中的流转信息
     */
    private TaskSupport getSupport(AsyncParam param) {
        return new TaskSupport().setParam(param).setRuleName(ruleName);
    }

    /**
     * 1、traceId
     * 2、日志
     *
     * @param support
     */
    private void logAdvance(TaskSupport support) {
        long traceId = IdWorker.nextId();
        TraceUtil.set(traceId);

        boolean costLogabled = ConfigManager.Action.costLogabled(ruleName);
        if (costLogabled) {
            LogWrapper.TimeCollector timeCollector = LogWrapper.TimeCollector.builder().startTime(SystemClock.now()).build();
            LogWrapper logWrapper = new LogWrapper().setTraceId(traceId).setTimeCollector(timeCollector);
            support.setLogWrapper(logWrapper);
            support.getTaskLoader().setLogWrapper(logWrapper);
        }
    }

    private void rule(String rulename) {
        this.ruleName = rulename;
    }

}
